package voldemort.store.cachestore.voldeimpl;

/**
 * Created by IntelliJ IDEA.
 * User: mhsieh
 * Date: 1/25/11
 * Time: 2:42 PM
 * To change this template use File | Settings | File Templates.
 */
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import voldemort.VoldemortException;
import voldemort.annotations.jmx.JmxGetter;
import voldemort.annotations.jmx.JmxManaged;
import voldemort.annotations.jmx.JmxOperation;
import voldemort.store.NoSuchCapabilityException;
import voldemort.store.StorageEngine;
import voldemort.store.StoreCapabilityType;
import voldemort.store.cachestore.BlockSize;
import voldemort.store.cachestore.CacheBlock;
import voldemort.store.cachestore.Key;
import voldemort.store.cachestore.Value;
import voldemort.store.cachestore.impl.CacheStore;
import voldemort.store.cachestore.impl.ChannelStore;
import voldemort.store.cachestore.impl.Purge;
import voldemort.store.cachestore.impl.PurgeThread;
import voldemort.utils.ClosableIterator;
import voldemort.utils.Pair;
import voldemort.utils.Props;
import voldemort.versioning.ClockEntry;
import voldemort.versioning.VectorClock;
import voldemort.versioning.Version;
import voldemort.versioning.Versioned;

import javax.xml.ws.Service;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

import static voldemort.store.cachestore.impl.CacheValue.createValue;

@JmxManaged(description = "Start and stop all stores.")
public class VoldeStore<K, V, T> implements StorageEngine<K, V, T> {

    private static Log logger = LogFactory.getLog(VoldeStore.class);

    //interface which encapture the logic of determine storage block size
    private BlockSize blockSize;

    // Channel store
    // current index, which use to support compaction operation
    private int curIndex ;
    // for voldermort interface
    private String storeName;
    // node id of storage
    private short nodeId;
    //public final static String FileNamePrefix = "cachestore";
    private CacheStore cacheStore;
    // delay write
    private boolean delay;
    // for store property
    private Props prop;
    // keep track of timestamp for each operation, pack backup and purge
    private List<Long> statList = new CopyOnWriteArrayList<Long>();
    // keep track of recent cache stats  and move to constructor
    private RecentCacheStats recentCacheStats ;
    private TimerTask flushTask;

    public VoldeStore(String storeName, BlockSize blockSize, String path, short nodeId) {
        this(storeName, blockSize, path, nodeId, false);
    }

    public VoldeStore(String storeName, BlockSize blockSize, String path, short nodeId, boolean delay) {
        this(storeName, blockSize, path, nodeId, delay, 0);

    }

     public VoldeStore(String storeName, BlockSize blockSize, String path, short nodeId, boolean delay, int mode) {
        this.storeName = storeName;
        this.blockSize = blockSize;
        this.nodeId = nodeId;
        this.delay = delay;
        // default to 0, cache store will figure which file to use based the content
        this.curIndex = 0;
        //this.list.add( open(getFileName(storeName, curIndex), map) );
        this.cacheStore = new CacheStore( path, blockSize, curIndex, storeName, delay, mode );
        for ( int i = 0; i < 6 ; i++) statList.add(0L);
        recentCacheStats = new RecentCacheStats(this);
        // hook shutdown process to close all file and flush all delay writes
        // Runtime.getRuntime().addShutdownHook( new Thread( new Shutdown()));
     }


    /**
     * no implementation yet
     */
    public void truncate() {
        logger.warn("No support for removeChannels");
    }

    @Override
    public boolean isPartitionAware() {
        return false;
    }


    /**
     * delegate the get to cachestore and return a wrapper for versiond
     * @param k
     * @return
     */
    private Versioned get(Key k) {
        Value value = cacheStore.get(k);
        if ( value == null ) return null;
        else return  makeVersioned4Block( value) ;
    }

    /**
     * create a versiond from Value of cachestore
     * @param value
     * @return Voldemort versiond
     */
    private Versioned makeVersioned4Block(Value value) {
        List<ClockEntry> list = new ArrayList<ClockEntry>(1);
        list.add(new ClockEntry(value.getNode(), value.getVersion()) ) ;
        VectorClock clock = new VectorClock(list, System.currentTimeMillis());
        return  new Versioned(value.getData(), clock);
    }

    /**
     *
     * @param key - defined by vlodermort interface
     * @return List<Versioned<V>> - Value defined by voldermort
     * @throws VoldemortException
     */
    public List<Versioned<V>> get(K key, T transforms) throws VoldemortException {
        Key k = Key.createKey( key);
        Versioned v = get( k);
        if ( v == null ) return new ArrayList<Versioned<V>>(0);
        else {
            List<Versioned<V>> list =  new ArrayList<Versioned<V>>(1);
            list.add(v);
            return list;
        }
    }

    /**
     * this is the real interface to channel store
     * @param key - Key class defined by storage
     * @param value - Value interface defined by storage
     */

    private void put(Key key, Value<byte[]> value) {
        cacheStore.put(key, value);

    }

    /**
     *
     * @param keys The keys to check for.
     * @return
     * @throws VoldemortException
     */
    public Map<K, List<Versioned<V>>> getAll(Iterable<K> keys, Map<K, T> transforms) throws VoldemortException {
        Map<K, List<Versioned<V>>> result =  new HashMap<K, List<Versioned<V>>> ( ((Collection<K>) keys).size());
        //Maps.newHashMapWithExpectedSize(((Collection<K>) keys).size());
        for(K key: keys) {
            List<Versioned<V>> value = get(key, transforms.get(key));
            if(!value.isEmpty())
                result.put(key, value);
        }
        return result;
    }

    /**
     *
     * @param key The key to use
     * @param value The value to store and its version.
     * @throws VoldemortException
     */
    public void put(K key, Versioned<V> value, T transforms) throws VoldemortException {
        Key k = Key.createKey(key);
        long version =  getVersion( (VectorClock) value.getVersion());
        // temporary log version info
        //logger.info("v "+value.toString() +" get "+version);
        short node = getNode((VectorClock) value.getVersion());
        if ( ! (value.getValue() instanceof byte[])) throw new VoldemortException("value "+ value.getValue().getClass().getName()+
                " must be byte[]");
        //using nodeId in the versioned
        Value v = new BlockValue( value.getValue(), version, node);
        put(k, v);
    }

    /**
     * return the nodeId which hold the version no
     * @param clock
     * @return
     */
    private short getNode(VectorClock clock) {
        if ( clock.getEntries().size() == 0 ) return (short) nodeId;
        else {
            // get the last one in the entry
            return clock.getEntries().get(clock.getEntries().size()-1).getNodeId() ;
        }
    }

    public CacheStore getCacheStore() {
        return cacheStore;
    }

    /**
     * get next version # from Vector clock
     * @param clock
     * @return
     */
    public static long getVersion(VectorClock clock) {
        long toReturn = clock.getMaxVersion();
        if ( clock.getEntries().size() > 1 )
            toReturn ++;
        return ( toReturn < 1 ? 1 : toReturn);
    }

    /**
     *
     * @param key The key to delete
     * @param version The current value of the key
     * @return
     * @throws VoldemortException
     */
    public boolean delete(K key, Version version) throws VoldemortException {
        Key k = Key.createKey( key);
        return cacheStore.remove( k);
    }

    public String getName() {
        return storeName;
    }

    /**
     *
      * @param threshold - MB/sec
     */
   @JmxOperation(description = "Pack data for Embedded Storage.")
    public void pack(int threshold) {
        statList.set(0, System.currentTimeMillis());
        cacheStore.pack( threshold);
        statList.set(1, System.currentTimeMillis());
    }

    public void pack() {
        pack( 0);
    }
    /**
     *
     * @param path
     * @param threshold - MB/seconds
     */
    @JmxOperation(description = "Backup data for Embeded Storage. output - MB/sec")
    public void backup(String path, int threshold){
        // convert it into MB /sec
        int t =  threshold * 1024 * 1024;
        statList.set(2, System.currentTimeMillis());
        if ( path == null ) {
            String p = prop.getString("backup.path", "./backup");
            cacheStore.backup( p, t);
        }
        else cacheStore.backup( path, t);
        statList.set(3, System.currentTimeMillis());

    }

    // purge interface
    private Purge purge;

    public void startTask(long period, Purge purge) {
        this.purge = purge;
        //start purge thread only period > 0
        if ( period > 0)
            cacheStore.startTask( period, purge);
    }

    private int writeBackThread = 0;
    private List<CacheStore.WriteBackThread> writeThreadList = null;

    public void startWriteThread(int no) {
        this.writeBackThread = no;
        writeThreadList = cacheStore.startWriteThread(no);
    }

    public void startPackThread(long period){
        cacheStore.startPackThread( period);
    }

    public void passStatList() {
        if ( cacheStore != null ) cacheStore.setStatList( statList);
        else logger.warn("CacheStore is null, can not pass statList");
    }

    /**
     * number of seconds
     * @param period
     */
    @JmxOperation(description = "Start flush task")
    public void startFlushThread(int period) {
        logger.info("flush thread run in minutes "+period);
        if ( flushTask != null ) flushTask.cancel() ;
        flushTask = new TimerTask() {
            @Override
            public void run() {
                try {
                    if ( cacheStore != null ) {
                        cacheStore.forceFlush();
                    }
                } catch (Exception ex) {
                    //swallow exception
                    logger.error( ex.getMessage(), ex);
                }
            }
        } ;
        new Timer().schedule( flushTask, period * 60*1000 , period * 60*1000);
    }

    @JmxOperation(description = "Purge data for storage.")
    public void purge() {
        statList.set(4, System.currentTimeMillis());
        if ( purge == null ) {
            logger.warn("no purge interface configured, process stop !!!");
        }
        else {
            new PurgeThread( purge, cacheStore ).run();
        }
        statList.set(5, System.currentTimeMillis());
    }

    @JmxOperation(description = "Get statistic from Embedded Storage.")
    public String getStoreStat() {
        return "Store "+ storeName+ " Node id "+nodeId+"\n"+cacheStore.getStat()+"\n" + getStatList() ;
    }

    @JmxGetter(name="Store Name")
    public String getStoreName(){ return storeName; }
    @JmxGetter(name="Node Id")
    public String getNodeId(){ return ""+nodeId; }
    @JmxGetter(name="File Name")
    public String getFileName(){ return cacheStore.getNamePrefix()+cacheStore.getCurIndex(); }
    @JmxGetter(name="Total Records")
    public long getTotalRecords(){ return cacheStore.getList().get(cacheStore.getCurIndex()).getTotalRecord(); }
    @JmxGetter(name="Total Deleted Records")
    public long getTotalDeletedRecords(){ return cacheStore.getDeleted(); }
    @JmxGetter(name="Total Active Records")
    public long getTotalActiveRecords(){ return (cacheStore.getList().get(cacheStore.getCurIndex()).getTotalRecord()-cacheStore.getDeleted()); }
    @JmxGetter(name="Total Active Records Percentage")
    public long getTotalActiveRecordsPercentage(){
        long hit = getTotalActiveRecords();
        long miss = getTotalDeletedRecords();
        return ( miss == 0 ? 100 : ( hit * 100 / (hit+miss) )); }
    @JmxGetter(name="File Size")
    public long getFileSize(){ return cacheStore.getList().get(cacheStore.getCurIndex()).getDataOffset(); }
    @JmxGetter(name="Block Overflow")
    public long getBlockOverflow(){ return cacheStore.getOverflow().get(); }
    @JmxGetter(name="Purge Trigger")
    public long getPurgeTrigger(){ return cacheStore.getTrigger(); }
    @JmxGetter(name="Cache Hits")
    public long getCacheHits(){ return cacheStore.getCacheHit().get(); }
    @JmxGetter(name="Cache Misses")
    public long getCacheMisses(){ return cacheStore.getCacheMis().get(); }
    @JmxGetter(name="Cache Hit Percentage")
    public long getCacheHitPercentage(){
        long hit = cacheStore.getCacheHit().get();
        long miss = cacheStore.getCacheMis().get();
        return ( miss == 0 ? 100 : ( hit * 100 / (hit+miss) )); }
    @JmxGetter(name="Recent Cache Hits")
    public long getRecentCacheHits(){ return recentCacheStats.getRecentCacheHits(); }
    @JmxGetter(name="Recent Cache Misses")
    public long getRecentCacheMisses(){ return recentCacheStats.getRecentCacheMisses(); }
    @JmxGetter(name="Recent Cache Hit Percentage")
    public long getRecentCacheHitPercentage(){
        long hit = recentCacheStats.getRecentCacheHits();
        long miss = recentCacheStats.getRecentCacheMisses();
        return ( miss == 0 ? 100 : ( hit * 100 / (hit+miss) )); }
    @JmxGetter(name="Pack Start Time")
    public String getPackStartTime(){ if ( statList.get(0) != 0 && statList.get(1) != 0)  return ""+toDate(statList.get(0)); else return "no record";  }
    @JmxGetter(name="Pack End Time")
    public String getPackEndTime(){ if ( statList.get(0) != 0 && statList.get(1) != 0)  return ""+toDate( statList.get(1)); else return "no record";  }
    @JmxGetter(name="Pack Duration")
    public String getPackDuration(){ if ( statList.get(0) != 0 && statList.get(1) != 0)  return ""+ (statList.get(1) - statList.get(0)); else return "no record";  }
    @JmxGetter(name="Backup Start Time")
    public String getBackupStartTime(){ if ( statList.get(2) != 0 && statList.get(3) != 0)  return ""+toDate(statList.get(2));  else return "no record"; }
    @JmxGetter(name="Backup End Time")
    public String getBackupEndTime(){ if ( statList.get(2) != 0 && statList.get(3) != 0)  return ""+toDate( statList.get(3));  else return "no record"; }
    @JmxGetter(name="Backup Duration")
    public String getBackupDuration(){ if ( statList.get(2) != 0 && statList.get(3) != 0)  return ""+ (statList.get(3) - statList.get(2));  else return "no record"; }
    @JmxGetter(name="Purge Start Time")
    public String getPurgeStartTime(){ if ( statList.get(4) != 0 && statList.get(5) != 0) return ""+toDate(statList.get(4)); else return "no record"; }
    @JmxGetter(name="Purge End Time")
    public String getPurgeEndTime(){ if ( statList.get(4) != 0 && statList.get(5) != 0) return ""+toDate( statList.get(5)); else return "no record"; }
    @JmxGetter(name="Purge Duration")
    public String getPuregeDuration(){ if ( statList.get(4) != 0 && statList.get(5) != 0)  return ""+ (statList.get(5) - statList.get(4)); else return "no record"; }
    @JmxGetter(name="Get current store info")
    public String getStoreInfo() {
        ChannelStore cs = cacheStore.getList().get( cacheStore.getCurIndex());
        return "key "+cs.getKeyOffset()+" data "+cs.getDataOffset();
    }
    private String getStatList() {
        StringBuilder sb = new StringBuilder();
        if ( statList.size() == 0) return "No records";
        else {
            sb.append("Pack Data ");
            if ( statList.get(0) != 0 && statList.get(1) != 0)
                sb.append(" begin "+toDate(statList.get(0)) + " end " + toDate( statList.get(1)) +
                sb.append(" duration "+ (statList.get(1) - statList.get(0)) ) );
            else sb.append(" no record");
            sb.append(", Back up Data ");
            if ( statList.get(2) != 0 && statList.get(3) != 0)
                sb.append(" begin "+toDate(statList.get(2)) + " end " + toDate( statList.get(3)) +
                sb.append(" duration "+ (statList.get(3) - statList.get(2) ) ) );
            else sb.append(" no record ");
            sb.append(", Purge Data ");
            if ( statList.get(4) != 0 && statList.get(5) != 0)
                sb.append(" begin "+toDate(statList.get(4)) + " end " + toDate( statList.get(5)) +
                sb.append(" duration "+ (statList.get(5) - statList.get(4)) ) );
            else sb.append(" no record");
            return sb.toString() ;
        }
    }


    @JmxGetter(name="Delay queue remain capacity")
    public int getDelayQueueCapacity() {

        if (cacheStore.getDelayWriteQueue() == null )
            return 0;
        else {
            return cacheStore.getDelayWriteQueue().remainingCapacity() ;
        }
    }

    @JmxGetter(name="Writeback threads and status")
    public String getWriteBackThread() {
        StringBuffer toReturn =  new StringBuffer("threads "+writeBackThread) ;
        if ( writeBackThread > 0  &&  writeThreadList != null ) {
            for ( int i = 0 ; i < writeBackThread ; i ++ ) {
                toReturn.append(" t "+i+" "+ writeThreadList.get(i).getStat()+" ");
            }
        }
        return toReturn.toString();
    }

    @JmxGetter(name="Recent Count")
    public long getRecentCount() {
        return recentCacheStats.getRecentCount();
    }
    @JmxGetter(name="Recent Skip")
    public long getRecentSkips() {
        return recentCacheStats.getRecentSkips();
    }
    @JmxGetter(name="Recent Empty")
    public long getRecentEmpty(){
        return recentCacheStats.getRecentEmpty();
    }


    private String toDate(long timestamp) {
        return new Date( timestamp).toString();
    }


    public void close() throws VoldemortException {
        cacheStore.close();
        if ( flushTask != null ) flushTask.cancel();
    }

    public Object getCapability(StoreCapabilityType capability) {
        throw new NoSuchCapabilityException(capability, getName());
    }

    /**
     *
     * @param key
     * @return
     */
    public List<Version> getVersions(K key) {
        List<Version> list = new ArrayList<Version>();
        Key k = Key.createKey( key);
        Versioned ver = get(k);
        // first version == 0 , when it is null
        if ( ver == null) list.add( new VectorClock());
        else list.add( ver.getVersion() );
        return list;
    }


    /**
     *  Nothing was done
      * @return Key value pair iterator
     */
   public ClosableIterator<Pair<K, Versioned<V>>> entries() {
        return new VoldeIterator( cacheStore.getMap());
    }


    /**
     *
      * @return Key iterator
     */
    public ClosableIterator<K> keys() {
        return new VoldeKeyIterator( cacheStore.getMap());
    }

    public void setProp(Props prop){
        this.prop = prop;
    }

    public class VoldeKeyIterator<K> implements  ClosableIterator<K>  {
            private final Iterator<Key> iterator;
            private Key currentKey;

            public VoldeKeyIterator(Map<Key, CacheBlock> map) {
                this.iterator = map.keySet().iterator() ;
            }

            @Override
            public void close() {
                // do nothing
            }

            @Override
            public boolean hasNext() {
                return iterator.hasNext();
            }

            @Override
            public K next() {
                while ( iterator.hasNext()) {
                    currentKey = iterator.next();
                    return (K) currentKey.getKey();
                }
                return null;
            }

            @Override
            public void remove() {
                // do nothing
            }
        }

    /**
     *
     * @param <K>
     * @param <V>
     */
    public class VoldeIterator<K, V> implements ClosableIterator<Pair<K, Versioned<V>>>  {

        private final Iterator<Map.Entry<Key, CacheBlock>> iterator;
        private Key currentKey;

        public VoldeIterator(Map<Key, CacheBlock> map){
            iterator = map.entrySet().iterator();
        }


        public void close() {
            //do nothing
        }

        public boolean hasNext() {
            return iterator.hasNext();
        }


        public Pair<K, Versioned<V>> next() {
            Versioned<V> versioned = null ;
            while ( iterator.hasNext() ) {
                Map.Entry<Key , CacheBlock> entry = iterator.next();
                currentKey = entry.getKey();
                CacheBlock block = entry.getValue();
                Value value ;
                try {
                    ChannelStore channel = cacheStore.getChannel( block);
                    byte[] data = channel.readChannel( block.getDataOffset2Len(), channel.getDataChannel());
                    value = createValue(data, block.getVersion() , block.getNode());
                    versioned = makeVersioned4Block( value);
                    // exit loop and return Pairs
                    break;
                } catch ( IOException ex) {
                    // swallow exception and move to next
                    logger.error(ex.getMessage(), ex);
                }
            }
            K key = (K) currentKey.getKey() ;
            return Pair.create( key, versioned );
        }

        public void remove() {
            //do nothing.
        }
    }


    /**
     * Class that implement channelstore's Value interface to bridge Voldermort versioned
     */
    class BlockValue implements Value<V> {
        V data;
        long version;
        short node;

        BlockValue(V data, long version, short node) {
            this.data = data;
            this.version = version;
            this.node = node;
        }

        public V getData() {
            return data;
        }

        public void setData(V data) {
            this.data = data;
        }

        public long getVersion() {
            return version;
        }

        public void setVersion(long version) {
            this.version = version;
        }

        public short getNode() {
            return node;
        }

        public void setNode(short node) {
            this.node = node;
        }
    }



    class RecentCacheStats{

        private final VoldeStore voldeStore;
        private Long millisecondsPerInterval = 5L;
        private Timer timer = new Timer();

        private long recentCacheHits;
        private long  recentCacheMisses;

        private long cumulativeCacheHits;
        private long cumulativeCacheMisses;

        // add writeThread
        private long recentSkips;
        private long recentEmpty;
        private long recentCount;
        private long cumSkips;
        private long cumEmpty;
        private long cumCount;

        public long getRecentCacheHits(){ return recentCacheHits; }
        public long getRecentCacheMisses(){ return recentCacheMisses; }

        public long getRecentSkips() {
            return recentSkips;
        }

        public long getRecentEmpty() {
            return recentEmpty;
        }

        public long getRecentCount() {
            return recentCount;
        }

        public RecentCacheStats(VoldeStore store){
            voldeStore = store;

            timer.schedule(
                         new TimerTask(){
                             public void run(){
                                 long currCacheHits = voldeStore.getCacheHits();
                                 long currCacheMisses = voldeStore.getCacheMisses();
                                 recentCacheHits = currCacheHits - cumulativeCacheHits;
                                 recentCacheMisses = currCacheMisses - cumulativeCacheMisses;
                                 cumulativeCacheHits = currCacheHits;
                                 cumulativeCacheMisses = currCacheMisses;
                                 if ( writeBackThread > 0  &&  writeThreadList != null ) {
                                     long skips =0 , empty = 0 , count = 0;
                                     for ( int i = 0 ; i < writeBackThread ; i ++ ) {
                                         skips += writeThreadList.get(i).getDirtyNo() ;
                                         empty += writeThreadList.get(i).getEmptyNo();
                                         count += writeThreadList.get(i).getCount();
                                     }
                                     recentSkips = skips - cumSkips;
                                     recentEmpty = empty - cumEmpty;
                                     recentCount = count - cumCount;

                                     cumSkips = skips;
                                     cumEmpty = empty;
                                     cumCount = count;
                                 } // if
                             } },
                         millisecondsPerInterval, millisecondsPerInterval
                 );
             }

        private long minutesToMilliseconds(long minutes){ return 60000*minutes; }
    }
}

